Crate rumqttc

A pure rust MQTT client which strives to be robust, efficient and easy to use. This library is backed by an async (tokio) eventloop which handles all the robustness and and efficiency parts of MQTT but naturally fits into both sync and async worlds as we'll see

Let's jump into examples right away

A simple synchronous publish and subscribe

use rumqttc::{MqttOptions, Client, QoS};
use std::time::Duration;
use std::thread;

let mut mqttoptions = MqttOptions::new("rumqtt-sync", "test.mosquitto.org", 1883);
mqttoptions.set_keep_alive(5);

let (mut client, mut connection) = Client::new(mqttoptions, 10);
client.subscribe("hello/rumqtt", QoS::AtMostOnce).unwrap();
thread::spawn(move || for i in 0..10 {
   client.publish("hello/rumqtt", QoS::AtLeastOnce, false, vec![i; i as usize]).unwrap();
   thread::sleep(Duration::from_millis(100));
});

// Iterate to poll the eventloop for connection progress
for (i, notification) in connection.iter().enumerate() {
    println!("Notification = {:?}", notification);
}

A simple asynchronous publish and subscribe

use rumqttc::{MqttOptions, AsyncClient, QoS};
use tokio::{task, time};
use std::time::Duration;
use std::error::Error;

let mut mqttoptions = MqttOptions::new("rumqtt-async", "test.mosquitto.org", 1883);
mqttoptions.set_keep_alive(5);

let (mut client, mut eventloop) = AsyncClient::new(mqttoptions, 10);
client.subscribe("hello/rumqtt", QoS::AtMostOnce).await.unwrap();

task::spawn(async move {
    for i in 0..10 {
        client.publish("hello/rumqtt", QoS::AtLeastOnce, false, vec![i; i as usize]).await.unwrap();
        time::delay_for(Duration::from_millis(100)).await;
    }
});

loop {
    let notification = eventloop.poll().await.unwrap();
    println!("Received = {:?}", notification);
    tokio::time::delay_for(Duration::from_secs(1)).await;
}

Quick overview of features

• Eventloop orchestrates outgoing/incoming packets concurrently and hadles the state
• Pings the broker when necessary and detects client side half open connections as well
• Throttling of outgoing packets (todo)
• Queue size based flow control on outgoing packets
• Automatic reconnections by just continuing the eventloop.poll()/connection.iter() loop`
• Natural backpressure to client APIs during bad network
• Immediate cancellation with client.cancel()

In short, everything necessary to maintain a robust connection

NOTE: Looping on connection.iter()/eventloop.poll() is necessary to run the eventloop and make progress. It yields incoming and outgoing activity notifications which allows customization as user sees fit.

IMPORTANT Blocking inside eventloop.poll()/eonnection.iter() loop will block connection progress

Since the eventloop is externally polled (with iter()/poll() in a loop) out side the library and Eventloop is accessible, users can

• Distribute incoming messages based on topics
• Stop it when required
• Access internal state for use cases like graceful shutdown or to modify options before reconnection

Structs

AsyncClient	AsyncClient to communicate with MQTT Eventloop This is cloneable and can be used to asynchronously Publish, Subscribe.
Client	Client to communicate with MQTT eventloop Connection.
ClientConfig	Common configuration for (typically) all connections made by a program.
ConnAck	Acknowledgement to connect packet
Connect	Connection packet initiated by the client
Connection	MQTT connection. Maintains all the necessary state
Disconnect
EventLoop	Eventloop with all the state of a connection
FixedHeader	Packet type from a byte
LastWill	LastWill that broker forwards on behalf of the client
Login
MqttOptions	Options to configure the behaviour of mqtt connection
MqttState	State of the mqtt connection.
Network	Network transforms packets <-> frames efficiently. It takes advantage of pre-allocation, buffering and vectorization when appropriate to achieve performance
PingReq
PingResp
PubAck	Acknowledgement to QoS1 publish
PubComp	Acknowledgement to pubrel
PubRec	Acknowledgement to QoS2 publish
PubRel	Acknowledgement to pubrec
Publish	Publish packet
PublishRaw	Raw publish is used to perform serialization in current thread rather the eventloop. Also client's eventloop stores serialized PublishRaw in state to prevent computation while retrying
SendError	An error returned from Sender::send().
Sender	The sending side of a channel.
SubAck	Acknowledgement to subscribe
Subscribe	Subscription packet
SubscribeTopic	Subscription filter
UnsubAck	Acknowledgement to unsubscribe
Unsubscribe	Unsubscribe packet

Enums

ClientError	Client Error
ConnectReturnCode	Return code in connack
ConnectionError	Critical errors during eventloop polling
Error
Event	Events which can be yielded by the event loop
Key	Key type for TLS authentication
Outgoing	Current outgoing activity on the eventloop
Packet	Encapsulates all MQTT packet types
PacketType	MQTT packet type
Protocol	Protocol type
QoS	Quality of service
Request	Requests by the client to mqtt event loop. Request are handled one by one. This is a duplicate of possible MQTT packets along with the ability to tag data and do bulk operations. Upcoming feature: When 'manual' feature is turned on provides the ability to reply with acks when the user sees fit
SecurityOptions	Client authentication option for mqtt connect packet
StateError
SubscribeReturnCodes	Subscription return code
TrySendError	An error returned from Sender::try_send().

Functions

certs	Extract all the certificates from rd, and return a vec of key::Certificates containing the der-format contents.
check	Checks if the stream has enough bytes to frame a packet and returns fixed header
has_wildcards	Checks if a topic or topic filter has wildcards
matches	Checks if topic matches a filter. topic and filter validation isn't done here.
mqtt_read	Reads a stream of bytes and extracts MQTT packets
pkcs8_private_keys	Extract all PKCS8-encoded private keys from rd, and return a vec of key::PrivateKeys containing the der-format contents.
qos	Maps a number to QoS
rsa_private_keys	Extract all RSA private keys from rd, and return a vec of key::PrivateKeys containing the der-format contents.
valid_filter	Checks if the filter is valid
valid_topic	Checks if a topic is valid

Type Definitions

Incoming